package avicit.bdp.dcs.stream.base;

import avicit.bdp.common.utils.enums.JobType;
import avicit.bdp.dcs.detail.dto.DataDetailDto;
import avicit.bdp.dcs.detail.dto.TableDetailDto;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.stream.dto.MeasureParaDto;
import avicit.bdp.dcs.stream.utils.Constants;
import avicit.bdp.dcs.stream.utils.TestUtils;
import avicit.bdp.dcs.tools.enums.MeasureParaTypeEnum;
import avicit.platform6.core.rest.msg.ResponseMsg;
import com.alibaba.fastjson2.JSON;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;

import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import static org.junit.Assert.assertEquals;

public class Kafka2IotdbTest {
    private RestTemplate restTemplate  = new RestTemplate();

    private String jobName = "Kafka2Iotdb门槛用例";

    private static String srcTble;

    private static String dstTble = "root.device.069829f0-4a3f-4200-8276-f1167931add0";

    private static String filePath = "device.txt";

    private static String hdfsPath = "hdfs://192.168.0.28:8020/testUnit";

    /**
     * 所有用例执行前准备工作
     */
    @BeforeClass
    public static void init() {
        System.out.println("Kafka2Iotdb门槛用例用例初始化...");
    }

    /**
     * 单个测试用例运行前准备工作
     */
    @Before
    public void setUp() {
    }

    /**
     * 单个测试用例运行后销毁工作
     */
    @After
    public void tearDown() {
    }

    /**
     * 所有用例执行完成后准备工作
     */
    @AfterClass
    public static void destory() {
        System.out.println("Kafka2Iotdb门槛用例执行完成，开始清理...");
    }

    /***********************************************Begin 测试用例 ***********************************************/
    /**
     */
    @Test(timeout = 5 * 60 * 1000)
    public void testKafka2Iotdb() throws Exception {
        // 获取模板列表
        ResponseEntity<ResponseMsg> result = restTemplate.getForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/stream" +
                        "/template/getMsgTemplateListAll",
                ResponseMsg.class);
        assertEquals("200", result.getBody().getRetCode());

        // 判断删除虚拟设备2模板
        List<Map> responseBody = (List<Map>) result.getBody().getResponseBody();
        for (Map msgTemplateDto: responseBody) {
            if (msgTemplateDto.get("deviceId").equals("069829f0-4a3f-4200-8276-f1167931add0")) {

                // 删除消息模板
                result = restTemplate.exchange(Constants.DCS_HOSTPORT + "/api/bdp/dcs/stream" +
                        "/template/deleteMsgTemplate", HttpMethod.DELETE,
                        TestUtils.buildDeleteMsgTemplateParas((String) msgTemplateDto.get("id")), ResponseMsg.class);
                assertEquals("200", result.getBody().getRetCode());
            }
        }

        // 新增模板
        result = restTemplate.postForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/stream/template/createMsgTemplate",
                TestUtils.buildSaveTemplateParas(), ResponseMsg.class);
        assertEquals("200", result.getBody().getRetCode());

        // 获取新增消息模板id
        // 获取模板列表
        result = restTemplate.getForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/stream" +
                        "/template/getMsgTemplateListAll",
                ResponseMsg.class);
        assertEquals("200", result.getBody().getRetCode());
        responseBody = (List<Map>) result.getBody().getResponseBody();
        for (Map msgTemplateDto: responseBody) {
            if (msgTemplateDto.get("deviceId").equals("069829f0-4a3f-4200-8276-f1167931add0")) {
                srcTble = (String) msgTemplateDto.get("id");
            }
        }

        // 新增参数
        MeasureParaDto measureParaDto = new MeasureParaDto();
        measureParaDto.setName("Device_Temperature");
        measureParaDto.setRemark("设备温度");
        measureParaDto.setType(MeasureParaTypeEnum.DOUBLE);
        result = restTemplate.postForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/stream/template/addMsgTemplateParas",
                TestUtils.buildaddMsgTemplateParas(measureParaDto, srcTble), ResponseMsg.class);
        assertEquals("200", result.getBody().getRetCode());

        measureParaDto.setName("Device_Pressure");
        measureParaDto.setRemark("设备压力");
        measureParaDto.setType(MeasureParaTypeEnum.DOUBLE);
        result = restTemplate.postForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/stream/template/addMsgTemplateParas",
                TestUtils.buildaddMsgTemplateParas(measureParaDto, srcTble), ResponseMsg.class);
        assertEquals("200", result.getBody().getRetCode());

        measureParaDto.setName("Device_Flow");
        measureParaDto.setRemark("设备流量");
        measureParaDto.setType(MeasureParaTypeEnum.DOUBLE);
        result = restTemplate.postForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/stream/template/addMsgTemplateParas",
                TestUtils.buildaddMsgTemplateParas(measureParaDto, srcTble), ResponseMsg.class);
        assertEquals("200", result.getBody().getRetCode());

        // step1:保存数据源
        result = restTemplate.postForEntity(Constants.DCS_HOSTPORT
                        + "/api/bdp/dcs/job/JobRest/saveDataSource",
                TestUtils.buildSaveDataSourceParas(JobType.STREAM.getCode(),
                jobName, Constants.KAFKA_DATASOURCE_ID, Constants.IOTDB_DATASOURCE_ID),
                ResponseMsg.class);
        assertEquals("200", result.getBody().getRetCode());

        // 获取创建的JobId
        ResponseMsg responseMsg = result.getBody();
        Map<String, String> jobMap = (Map<String, String>)responseMsg.getResponseBody();
        String jobId = jobMap.get("jobId");

        try {
            // step2:保存表映射关系
            result = restTemplate.postForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/job/JobRest/saveTableMapping",
                    TestUtils.buildSaveTblMapping(JobType.STREAM.getCode(), jobId, srcTble, dstTble), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());

            // 10：获取任务信息
            result = restTemplate.getForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/task/TaskRest/getTaskById?jobId=" +
                            jobId + "&srcTableName=" + "虚拟设备2",
                    ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            System.out.println("10：获取任务信息成功");
            Map taskMap = (Map) result.getBody().getResponseBody();
            // 获取任务id
            String taskId = (String) taskMap.get("id");

            // 11：修改任务配置
            result = restTemplate.postForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/task/TaskRest/updateTask",
                    TestUtils.buildUpdateIotdbTaskParam(taskId, srcTble, dstTble), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            System.out.println("11：修改任务配置成功");

            // step3:任务配置检查
            result = restTemplate.postForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/job/JobRest/checkTaskConfig",
                    TestUtils.buildCheckTaskConfig(jobId), ResponseMsg.class);
            assertEquals(true, result.getBody().getResponseBody());

            // step4:预检查
            result = restTemplate.postForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/precheck/PrecheckRest" +
                            "/executePrecheck",
                    TestUtils.buildPreChecks(jobId), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());

            // step5:更新Job状态
            result = restTemplate.postForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/job/JobRest/updateJobStatus",
                    TestUtils.buildUpdateJobStatus(jobId), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());

            // step6:启用/禁用Job

            // step7:启动任务
            result = restTemplate.postForEntity(Constants.DCS_HOSTPORT + "/api/bdp/dcs/job/JobRest/startJobByIds",
                    TestUtils.buildStartJob(jobId), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());

            // step8:获取Job列表
            // 等待DS启动任务成功
            Thread.sleep(50 * 1000);
            result = restTemplate.exchange(Constants.DCS_HOSTPORT + "/api/bdp/dcs/job/JobRest/getJobList" +
                            "?jobStatus&sort&name=Kafka2Iotdb门槛用例&jobType=5&currentPage=1&pageSize=10",
                    HttpMethod.GET, TestUtils.buildGetHttpEntity(), ResponseMsg.class);
            ResponseMsg body = result.getBody();
            LinkedHashMap<String, Object> map = (LinkedHashMap<String, Object>)body.getResponseBody();
            List<Object> finalBody = (List<Object>) map.get("result");
            for (Object obj : finalBody) {
                LinkedHashMap<String, String> jobObj = (LinkedHashMap<String, String>)obj;
                JobDTO jobInfo = JSON.parseObject(JSON.toJSONString(jobObj), JobDTO.class);
                if (jobInfo.getId().equals(jobId)) {
                    System.out.println("Kafka2Iotdb门槛用例任务状态:" + jobInfo.getJobStatus());

                    // 判断任务是否同步完成;
                    assertEquals(1 == jobInfo.getJobStatus(), true);

                    // 判断同步速度;（备注：源表中只有一行数据，大小固定）
//                    assertEquals(jobIfo.getReadSucceedBytes() == 45, true);
                    break;
                }
            }

            // step9:查看日志信息
            result = restTemplate.exchange(Constants.DCS_HOSTPORT + "/api/bdp/dcs/log/LogRest/getLogByTaskId" +
                    "?taskInstanceId=&skipLineNum=0" +
                    "&limit=100&jobId="+jobId, HttpMethod.GET, TestUtils.buildGetHttpEntity(), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());

            // step10:查看详情，包括表结构迁移、数据迁移；
            result = restTemplate.exchange(Constants.DCS_HOSTPORT + "/api/bdp/dcs/detail/TaskTableDetailRest" +
                            "/getTaskTableDetailList" +
                            "?currentPage=1&pageSize=20&srcTableName=&jobId="+jobId, HttpMethod.GET,
                    TestUtils.buildGetHttpEntity(), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());

            body = result.getBody();
            map = (LinkedHashMap<String, Object>)body.getResponseBody();
            finalBody = (List<Object>) map.get("result");
            for (Object obj : finalBody) {
                LinkedHashMap<String, String> jobObj = (LinkedHashMap<String, String>)obj;
                TableDetailDto tableDetailDto = JSON.parseObject(JSON.toJSONString(jobObj), TableDetailDto.class);
                // 判断任务是否task是否同步完成;
                assertEquals(7 == tableDetailDto.getTaskStatus(), true);
            }

            result = restTemplate.exchange(Constants.DCS_HOSTPORT + "/api/bdp/dcs/detail/TaskDataDetailRest" +
                            "/getTaskDataDetailList" +
                            "?currentPage=1&pageSize=20&srcTableName=&jobId="+jobId, HttpMethod.GET,
                    TestUtils.buildGetHttpEntity(),
                    ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            body = result.getBody();
            map = (LinkedHashMap<String, Object>)body.getResponseBody();
            finalBody = (List<Object>) map.get("result");
            for (Object obj : finalBody) {
                LinkedHashMap<String, String> jobObj = (LinkedHashMap<String, String>)obj;
                DataDetailDto dataDetailDto = JSON.parseObject(JSON.toJSONString(jobObj), DataDetailDto.class);
                // 判断任务是否task是否同步完成
                assertEquals(1 == dataDetailDto.getTaskStatus(), true);

                // 判断已完成数是否为3条
//                assertEquals(dataDetailDto.getCompletedTotal() == 3, true);
            }
        } finally {
//             step11: 删除job
            result = restTemplate.exchange(Constants.DCS_HOSTPORT + "/api/bdp/dcs/job/JobRest/deleteJobByIds",
                    HttpMethod.DELETE,
                    TestUtils.buildDeleteJob(jobId),
                    ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
        }
    }

}
